package cn.pugle.base.net.socket;

import cn.pugle.base.net.socket.client.CONS;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 使用NIO的Server 非阻塞， Reactor模式 = Selector
 * Reactor 仍是同步IO。 Proactor异步 事件回调
 * Selector(感觉有两个地方涉及用不用selector来多路复用，一个是ServerSocket accept的时候，一个是接收到的SocketChannel读数据的时候 )
 * 这版是 单Selector 单线程版
 * //todo Selector 是如何判断各个事件的？ 什么样算是可读？
 * //xxx 多线程的问题在与把同一个线程的读事件发给多个线程处理怎么办？
 * //ooo 读数据(IO) 和处理数据(CPU) 分离 写数据
 *
 * @author tzp
 * @date 15-12-11
 */
public class OneReactorServer {

    private InetSocketAddress inetAddress;
    private ExecutorService threadPool;
    private Selector selector;
    ByteBuffer receivebuffer = ByteBuffer.allocate(5);

    public OneReactorServer() {
        inetAddress = new InetSocketAddress(CONS.SERVER_ADDRESS, CONS.SERVER_PORT);
        threadPool = Executors.newFixedThreadPool(10);
        try {
            selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new OneReactorServer().bootstrap();
    }

    public void bootstrap() {
        ServerSocketChannel ssc = null;
        try {
            ssc = ServerSocketChannel.open();
            ssc.bind(inetAddress);
            ssc.configureBlocking(false);

            SelectionKey selectionKey = ssc.register(selector, SelectionKey.OP_ACCEPT);//向主Selector注册 ServerChannel，关心连接accept状态
            System.out.println(selector.getClass().toString());
            System.out.println("Reactor register ACCEPT with a socketChannel " + ssc.getLocalAddress().toString());
            dispatchLoop();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 之前出现一个问题就是接受连接后一直打印Reactor select a key, READ。 怀疑是SK没从set里清除。
     * 原来用for each循环 selectionKeys，然后clear.后来改成iterator.remove，其实效果是一样的，但是自己骗自己
     * 后来怀疑问题不是出现在这里没有从SK_Set中清除sk，而是handleSK启动的worker线程 没有处理可读的数据就一直可读
     * 最后发现 一直打印READ是发生在Client关闭之后，read()返回-1 ，竟然一直是可读
     */
    public void dispatchLoop() {
        try {
            while (!Thread.interrupted()) {
                selector.select();//阻塞的
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey sk = iterator.next();
                    iterator.remove();
                    handleSK(sk);
                }
//                selectionKeys.clear();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void handleSK(SelectionKey sk) throws IOException {
        if (sk.isAcceptable()) {
            System.out.println("Reactor select a key, ACCEPT");//sk 是int位
            ServerSocketChannel serverChannel = (ServerSocketChannel) sk.channel();//SelectionKey.accept()返回的是原来的ssk的serverSocket

            SocketChannel channel = serverChannel.accept();
            channel.configureBlocking(false);
            channel.register(selector, SelectionKey.OP_READ);
            System.out.println("Reactor register READ with a socketChannel " + channel.getRemoteAddress() + channel.getLocalAddress());
        } else if (sk.isReadable()) {
            System.out.println("Reactor select a key, READ");
            SocketChannel channel = (SocketChannel) sk.channel();
            receivebuffer.clear();
            int count = channel.read(receivebuffer);
            System.out.println(count + " bytes read");
            if (count > 0) {
//                new Thread(new MyChannelReader(channel, receivebuffer, count)).start();
                String receiveText = new String(receivebuffer.array(), 0, count);
                System.out.println("服务器端接受客户端数据: " + receiveText);
            } else if (count == -1) {
                sk.cancel();
                System.out.println(channel.isConnected());
                channel.close();//不关闭 会CLOSE_WAIT
                System.out.println("Reactor cancel a channel");
            }
        }
    }

    class MySocketChannelWorker implements Runnable {
        private SocketChannel channelToTreat;

        public MySocketChannelWorker(SocketChannel channelToTreat) {
            this.channelToTreat = channelToTreat;
        }

        @Override
        public void run() {
            try {
                long id = Thread.currentThread().getId();
                System.out.println(id + "begin treat socketChannel, remoteAddress: " + channelToTreat.getRemoteAddress().toString());
                ByteBuffer buffer = ByteBuffer.allocate(2 * 1024 * 1024);
                int readNum = channelToTreat.read(buffer);
                System.out.println(id + "+ " + readNum + " bytes read");
                StringBuilder stringBuilder = new StringBuilder();
                while (stringBuilder.lastIndexOf(CONS.PROTOCOL_TAIL) == -1 && readNum != -1) {
//                System.out.println("- its not blocking, you can do something else");
//                Thread.sleep(50 * 1000);
                    int curPos = buffer.position();
                    System.out.println(id + "- " + curPos + " current position");
                    buffer.flip();
                    byte[] dst = new byte[curPos];
                    buffer.get(dst);
                    String strRece = new String(dst, CONS.CHARSET_NAME);
                    System.out.println(id + "- receive: " + strRece);
                    stringBuilder.append(strRece);
                    buffer.clear();
                    //xxx 这两行代码之前出现在上面注释掉的地方。体现了对非阻塞过程的不了解，其实上次调用read已经彻彻底底的返回了，buffer里的数据就是要处理的了
                    //非阻塞的IO 只是系统在读到kernel里，read的时候拷贝到buffer里
                    System.out.println(id + "- its not blocking, you can do something else");
                    Thread.sleep(50 * 1000);
//                readNum = channelToTreat.read(buffer);
                }
                System.out.println(id + "+ end treat" + channelToTreat.getRemoteAddress().toString());
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    class MyBufferWorker implements Runnable {
        private SocketChannel channelToTreat;
        private ByteBuffer bufferToTreat;
        private int lenth;

        public MyBufferWorker(SocketChannel channel, ByteBuffer bufferToTreat, int lenth) {
            this.channelToTreat = channel;
            this.bufferToTreat = bufferToTreat;
            this.lenth = lenth;
        }

        @Override
        public void run() {
            String receiveText = new String(bufferToTreat.array(), 0, lenth);
            System.out.println("服务器端接受客户端数据--:" + receiveText);
            bufferToTreat.clear();
//                channelToTreat.register(selector, SelectionKey.OP_READ);

        }
    }
}



